package com.kerrykidz.system.util.sync;

import java.io.Serializable;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;

import org.apache.commons.lang3.SerializationUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

@Component
public class ConsumerTask {

	@Autowired
	@Qualifier("syncDataConsumer")
	private SyncConsumer syncDataConsumer;

	@Autowired
	@Qualifier("syncFileConsumer")
	private SyncConsumer syncFileConsumer;

	public void consumer() throws JMSException {
		listenerDataSync();
		listenerFileSync();
	}

	private void listenerDataSync() throws JMSException {

		Connection con = ActiveMQHelper.getConnection();
		Session session = con.createSession(true, Session.AUTO_ACKNOWLEDGE);
		Destination destination = session.createQueue(SyncParamHelper.CURRENT_STORE);
		final MessageConsumer producer = session.createConsumer(destination);
		new Thread() {
			public void run() {

				try {
					producer.setMessageListener(new MessageListener() {

						@Override
						public void onMessage(Message msg) {
							BytesMessage message = (BytesMessage) msg;
							try {
								if (message != null && !message.getJMSRedelivered()) {
									byte[] bys = new byte[(int) message.getBodyLength()];
									message.readBytes(bys);
									MessageTask task = (MessageTask) SerializationUtils.deserialize(bys);
									syncDataConsumer.doConsumer(task);
									message.acknowledge();
								}
							} catch (JMSException e) {
								e.printStackTrace();
							}
						}
					});
				} catch (JMSException e1) {
					e1.printStackTrace();
				}

			};
		}.start();

	}

	private void listenerFileSync() throws JMSException {

		Connection con = ActiveMQHelper.getConnection();
		Session session = con.createSession(true, Session.AUTO_ACKNOWLEDGE);
		Destination destination = session.createQueue(SyncParamHelper.CURRENT_FILE_QUEUE);
		final MessageConsumer producer = session.createConsumer(destination);
		new Thread() {
			public void run() {

				try {
					producer.setMessageListener(new MessageListener() {

						@Override
						public void onMessage(Message msg) {
							BytesMessage message = (BytesMessage) msg;
							try {
								if (message != null && !message.getJMSRedelivered()) {
									byte[] bys = new byte[(int) message.getBodyLength()];
									message.readBytes(bys);
									syncFileConsumer.doConsumer((Serializable) SerializationUtils.deserialize(bys));
								}
							} catch (JMSException e) {
								e.printStackTrace();
							}
						}
					});
				} catch (JMSException e1) {
					e1.printStackTrace();
				}

			};
		}.start();

	}
}
